package com.junxonline.instant.api.zookeeper;

import cn.hutool.core.date.DateUtil;
import com.junxonline.instant.api.config.ZookeeperConfig;
import lombok.extern.log4j.Log4j2;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.net.InetAddress;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Zookeeper Client
 *
 * @author JunX
 * @date 2021-09-08
 */
@Log4j2
@Component
public class ZookeeperClient {

    @Autowired
    private ZookeeperConfig zookeeperConfig;

    /**
     * Zookeeper客户端
     */
    private CuratorFramework zookeeperClient;

    /**
     * 计数器（线程安全）
     */
    private final AtomicInteger nextServerCyclicCounter = new AtomicInteger(0);

    /**
     * ServerNode容器（线程安全）
     */
    // key="/services/xxx/xxx-server-0000000001"  value="192.168.2.17:7021"
    // key="/services/xxx/xxx-server-0000000002"  value="192.168.2.17:7022"
    // key="/services/xxx/xxx-server-0000000003"  value="192.168.2.17:7023"
    private final ConcurrentMap<String, String> serverNode = new ConcurrentHashMap<>();

    /**
     * Zookeeper Client启动
     * @throws Exception Exception
     */
    @PostConstruct
    public void start() throws Exception {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(zookeeperConfig.getBaseSleepTimeMs(), zookeeperConfig.getMaxRetries());
        zookeeperClient = CuratorFrameworkFactory
                .builder()
                .connectString(zookeeperConfig.getZookeeperAddress())
                .retryPolicy(retryPolicy)
                .sessionTimeoutMs(zookeeperConfig.getSessionTimeoutMs()).connectionTimeoutMs(zookeeperConfig.getConnectionTimeoutMs()).build();
        zookeeperClient.start();

        String name = zookeeperConfig.getId();
        String zookeeperAddress = zookeeperConfig.getZookeeperAddress();
        String ipPort = InetAddress.getLocalHost().getHostAddress() + ":" + zookeeperConfig.getPort();
        String data = name + "#" + zookeeperAddress + "#" + ipPort;
        zookeeperClient.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(zookeeperConfig.getRegistryPath() + "/" + zookeeperConfig.getId() + "-", data.getBytes());
        log.info("- Zookeeper client started on -> {} in path: {}", ipPort, zookeeperConfig.getRegistryPath());

        // 对Zookeeper当中指定的Path进行监听
        CuratorCache curatorCache = CuratorCache.builder(zookeeperClient, zookeeperConfig.getListeningPath()).build();
        PathChildrenCacheListener pathChildrenCacheListener = (client, event) -> {
            event.getType();
            if (event.getType() == PathChildrenCacheEvent.Type.CONNECTION_RECONNECTED) {
                return;
            }
            log.info("Zookeeper event for basePath=[{}] event=[{}]", zookeeperConfig.getZookeeperAddress(), event);

            // 节点上线事件通知
            if (event.getType() == PathChildrenCacheEvent.Type.CHILD_ADDED) {
                String fullPath = event.getData().getPath();
                String eventData = new String(event.getData().getData(), StandardCharsets.UTF_8);
                log.info("Server node is online -> name: {}, info: {}", fullPath, eventData);
                serverNode.put(fullPath, eventData);
            }

            // 节点下线事件通知
            if (event.getType() == PathChildrenCacheEvent.Type.CHILD_REMOVED) {
                String fullPath = event.getData().getPath();
                log.info("Server node is offline -> name: {}", fullPath);
                serverNode.remove(fullPath);
            }
        };
        CuratorCacheListener listener = CuratorCacheListener.builder().forPathChildrenCache(zookeeperConfig.getListeningPath(), zookeeperClient, pathChildrenCacheListener).build();
        curatorCache.listenable().addListener(listener);
        curatorCache.start();
        log.info("Zookeeper client listen on path -> {}", zookeeperConfig.getListeningPath());
    }


    /**
     * Zookeeper Client停止
     */
    @PreDestroy
    public void stop() {
        zookeeperClient.close();
        log.info("Zookeeper client stopped");
    }

    /**
     * 轮询返回可用Server节点（线程安全）
     *
     * @return Object
     */
    public String getServerNode() {
        String node;
        int count = 0;

        // 循环10次
        while (count++ < 10) {
            Object[] keys = serverNode.keySet().toArray();
            int nodeCount = keys.length;

            if ((nodeCount == 0)) {
                log.warn("No up Server Node available");
                return null;
            }

            // 轮询获取数组下标（线程安全）
            int nextNodeIndex = incrementAndGetModulo(nodeCount);

            // 这里有可能返回空(因为在此刻有可能Zookeeper当中有事件发生导致ServerNode删除了某个节点)
            node = serverNode.get(keys[nextNodeIndex]);

            // 如果node为空说明刚好有节点掉线被删除
            if (node == null) {
                // 让出一会CPU资源
                Thread.yield();
                // 继续循环10次
                continue;
            }
            log.debug("Choose server node is [{}]", node);
            // 返回可用节点
            String[] dataArray = node.split("#");
            if (dataArray.length > 2) {
                return dataArray[2];
            }
        }

        if (count >= 10) {
            log.warn("No available alive servers after 10 tries from zookeeper");
        }

        return null;
    }

    /**
     * 根据一个模数轮询下标值
     *
     * @param modulo 模数(数组的长度)
     * @return 假如模数是5, 那么返回值为 0,1,2,3,4,0,1,2,3,4
     */
    private int incrementAndGetModulo(int modulo) {
        for (; ; ) {
            int current = nextServerCyclicCounter.get();
            // 拿出当前nextServerCyclicCounter的值+1 对 modulo 取模
            // 计算出的值next必定小于modulo
            int next = (current + 1) % modulo;
            // 乐观锁设值 如果nextServerCyclicCounter当前值还等于current则设值成功, 否则死循环解决并发问题
            if (nextServerCyclicCounter.compareAndSet(current, next)) {
                return next;
            }
        }
    }

}
